package com.rocketmq.demo;

import com.rocketmq.constants.MqConstants;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.junit.Test;

import java.util.Date;

/**
 * @ClassName DMsTest
 * @Description 延时消息
 * @Author Patrick Star
 * @Date 2025/7/27 16:43
 */
public class DMsTest {

    @Test
    public void msProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("ms-producer-group");
        producer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);
        producer.start();
        Message message = new Message("msTopic", "我是一个延时消息".getBytes());
        // 给消息设置一个延时时间 这里不是设置延时的时间 而是设置延时的level
        message.setDelayTimeLevel(2);
        // 发送延时消息
        producer.send(message);
        System.out.println("发送时间" + new Date());
        producer.shutdown();
    }

    /**
     * @throws Exception
     */
    @Test
    public void msConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ms-consumer-group");
        consumer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);
        consumer.subscribe("msTopic", "*");
        consumer.registerMessageListener((MessageListenerConcurrently) (message, context) -> {
            System.out.println("收到消息了" + new Date());
            System.out.println(message);
            System.out.println(new String(message.get(0).getBody()));
            System.out.println(context);
            // 返回收到消息成功了 就会将消息消费掉
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.in.read();
    }
}


